package hbec.kettle.etl.engine.schedule;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;

import hbec.kettle.etl.common.constants.DataBase;
import hbec.kettle.etl.engine.conf.EngineConf;
import hbec.kettle.etl.engine.conf.JobConf;
import hbec.kettle.etl.engine.conf.JobEntityConf;
import hbec.kettle.etl.engine.conf.PluginConf;
import hbec.kettle.etl.engine.conf.ReaderConf;
import hbec.kettle.etl.engine.conf.WriterConf;
import hbec.kettle.etl.engine.plugin.BufferedLineExchanger;
import hbec.kettle.etl.engine.plugin.Line;
import hbec.kettle.etl.engine.plugin.LineExchanger;
import hbec.kettle.etl.engine.storage.StoragePool;

public class Engine {
	private static final Logger logger = Logger.getLogger(Engine.class);
	
	private EngineConf engineConf;
	//private Map<String, PluginConf> pluginReg;
	//private MonitorPool readerMonitorPool;
	//private MonitorPool writerMonitorPool;
	
	public static Set<String> workingJob = new CopyOnWriteArraySet<>();
	private static Map<String, JobEntityConf> object2Job = new ConcurrentHashMap<>();
	
	private Map<DataBase, String> checkPointTableOfDb = new HashMap<>();
	
	public Engine(EngineConf engineConf, Map<String, PluginConf> pluginReg) {
		this.engineConf = engineConf;
		//this.pluginReg = pluginReg;

		//this.writerMonitorPool = new MonitorPool();
		//this.readerMonitorPool = new MonitorPool();
	}
	
	public void start(JobConf jobConf) throws Exception {
		logger.info("Engine startups.");
		
		for (String checkPointTable : engineConf.getCheckPointTable().split(",")) {
			checkPointTableOfDb.put(DataBase.valueOf(checkPointTable.split("\\.")[0].trim().toUpperCase()), checkPointTable.split("\\.")[1].trim());
		}
		
		StoragePool storagePool = new StoragePool();
		NamedThreadPoolExecutor readerPool = initReaderPool(jobConf, storagePool);
		readerPool.shutdown();
		
		NamedThreadPoolExecutor writerPool = initWriterPool(storagePool);
		while (true) {
			if (writerPool.getActiveCount() < 10) {
				Line line = LineExchanger.getFromReader();
				if (line != null && !workingJob.contains(line.getObject())) {
					if (workingJob.contains(line.getObject())) {
						if (BufferedLineExchanger.lineBuffer(line.getObject(), line)) 
							LineExchanger.popLine(line);
						else 
							Thread.sleep(3 * 1000);
					} else {
						line = BufferedLineExchanger.getLineBuffer(line.getObject(), line);
						workingJob.add(line.getObject());
						LineExchanger.popLine(line);
						
						WriterConf writerConf = new WriterConf();
						OK:
						if (!object2Job.containsKey(line.getObject())) {
							for (JobEntityConf jec : jobConf.getJobEntityConfs()) {
								for (String object : jec.getWatchTables()) {
									if (object.equalsIgnoreCase(line.getObject())) {
										object2Job.put(line.getObject(), jec);
										break OK;
									} 
								}
							}
						}
						
						Map<DataBase, String> watchSumTablesOfDb = new HashMap<>();
						for (String watchSumTable : engineConf.getWatchSumTables().split(",")) {
							watchSumTablesOfDb.put(DataBase.valueOf(watchSumTable.split("\\.")[0].trim().toUpperCase()), watchSumTable.split("\\.")[1].trim());
						}
						
						writerConf.setWatchSumTablesOfDb(watchSumTablesOfDb);
						writerConf.setJob(object2Job.get(line.getObject()));
						writerConf.setCheckPointTableOfDb(checkPointTableOfDb);
						WriterWorker writerWorker = new WriterWorker(writerConf, line);
						writerPool.execute(writerWorker);
					}
					Thread.sleep(1000);
				} else 
					Thread.sleep(3 * 1000);
			} else 
				Thread.sleep(10 * 1000);
		}
	}
	
	private NamedThreadPoolExecutor initReaderPool(JobConf jobConf, StoragePool storagePool) throws Exception {
		NamedThreadPoolExecutor readerPool = new NamedThreadPoolExecutor("reader", 2, 2, 
				0L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
		readerPool.prestartAllCoreThreads();
		
		for (String watchSumTable : engineConf.getWatchSumTables().split(",")) {
			ReaderConf readerConf = new ReaderConf();
			
			Map<DataBase, String> watchSumTablesOfDb = new HashMap<>();
			watchSumTablesOfDb.put(DataBase.valueOf(watchSumTable.split("\\.")[0].trim().toUpperCase()), watchSumTable.split("\\.")[1].trim());

			readerConf.setWatchSumTablesOfDb(watchSumTablesOfDb);
			readerConf.setCheckPointTableOfDb(checkPointTableOfDb);
			
			ReaderWorker readerWorker = new ReaderWorker(readerConf);
			readerPool.execute(readerWorker);
			//readerMonitorPool.monitor(readerWorker);
		}
		
		return readerPool;
	}
	
	private NamedThreadPoolExecutor initWriterPool(StoragePool storagePool) throws Exception {
		NamedThreadPoolExecutor writerPool = new NamedThreadPoolExecutor("writer", 4, 10, 
				5L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
		writerPool.prestartAllCoreThreads();
		
		return writerPool;
	}
}
